/**
 * 
 */
package com.rrd.dw.mr.gzip;

import java.io.IOException;
// import java.util.HashMap;
// import java.util.Map;
import java.io.InputStream;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.MD5Util;

/**
 * @author xiexihao
 */
public class CreditReportsExtractor extends Configured implements Tool {
    private static final String STR_FILED_SPLIT_FLAG = "\001";
    private static final String STR_TB_PREFIX_FLAG = "\007";

    public static final int SIZEOF_MD5 = 16;

    private static final byte[] FAMILY_INF = Bytes.toBytes("inf");
    private static final byte[] FAMILY_JOB = Bytes.toBytes("job");
    private static final byte[] FAMILY_DET = Bytes.toBytes("det");

    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_INF = "creditreportsinfo";
    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_JOB_INF = "creditreportsjobinfo";
    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF = "creditreportsexecinfo";
    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF = "creditreportsguarinfo";
    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF = "creditreportscredinfo";
    private static final String STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF = "creditreportsquerinfo";

    private static class CreditReportsMapper extends TableMapper<Text, Text> {
        private Text key = null;
        private Text value = null;

//    	private static Map<String, Map<String, byte[]>> map = null;

        protected void setup(Context context) {
            key = new Text();
            value = new Text();

//    		map = new HashMap<String, Map<String, byte[]>>();
        }

        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
            String keyStr = null;
            StringBuffer keysb = new StringBuffer();

            Object[] objs = decodeRowkey(key.get());
            String idCardNo = null;
            if (objs != null && objs[1] != null) {
                idCardNo = objs[1].toString();
                keysb.append(idCardNo);
            }
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("report_no")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("report_no"))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("cert_no")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("cert_no"))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("report_time")) == null
                    ? ""
                    : Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("report_time"))));
            keysb.append(value.getValue(FAMILY_DET,
                Bytes.toBytes("exec_case_no")) == null ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("exec_case_no"))));
            keysb.append(value.getValue(FAMILY_DET,
                Bytes.toBytes("exec_close_time")) == null ? ""
                    : Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("exec_close_time"))));
            keysb.append(value.getValue(FAMILY_DET,
                Bytes.toBytes("guar_cert_no")) == null ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("guar_cert_no"))));
            keysb.append(
                value.getValue(FAMILY_DET, Bytes.toBytes("quer_no")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("quer_no"))));
            keysb.append(
                value.getValue(FAMILY_DET, Bytes.toBytes("quer_time")) == null
                    ? ""
                    : Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("quer_time"))));
            keysb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))));
            keysb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null
                    ? ""
                    : Bytes.toLong(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("end_time"))));
            keyStr = MD5Util.getMd5(keysb.toString());
            this.key.set(keyStr);

            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_CREDIT_REPORTS_JOB_INF
                + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("system_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("successful")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("successful")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("type")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("type"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("try_times")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("try_times")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("requested")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("requested")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("query")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("query"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("start_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("start_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("timestamp")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("timestamp")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("finished")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("finished")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("message")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("message"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("data_file")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("data_file"))));

            this.value.set(sb.toString());

            context.write(this.key, this.value);
            parseBasicInfo(context, keyStr, value, idCardNo);
            parseDetInfo(context, keyStr, value);
        }

        private void parseBasicInfo(Context context, String keyStr,
                Result value, String idCardNo)
                throws IOException, InterruptedException {
            StringBuffer sb = new StringBuffer();
            sb.append(
                STR_TARGET_POSITION_CREDIT_REPORTS_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);

            sb.append(idCardNo).append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("report_no")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("report_no"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("cert_type")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("cert_type"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("cert_no")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("cert_no"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("real_name")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("real_name"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("married")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("married"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("gender")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("gender"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("request_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("request_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("report_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("report_time")))));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);
        }

//    	@SuppressWarnings("deprecation")
        private void parseDetInfo(Context context, String keyStr, Result value)
                throws IOException, InterruptedException {
            StringBuffer sb = new StringBuffer();
            Cell countCell = null;
            int count = 0;

            countCell = value.getColumnLatestCell(FAMILY_DET,
                Bytes.toBytes("exec_count"));
            if (countCell == null)
                System.out.println("---------exec--------");
            if (countCell != null) {
                count = Bytes.toInt(CellUtil.cloneValue(countCell));

                for (int i = 0; i < count; i++) {
                    sb.delete(0, sb.length());
                    sb.append(STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF
                        + STR_TB_PREFIX_FLAG);
                    sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i).append(STR_FILED_SPLIT_FLAG);

                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_case_no#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_case_no#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_court#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_court#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_reason#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_reason#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_over_way#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_over_way#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_filing_time#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("exec_filing_time#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_status#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_status#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_apply_exec#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_apply_exec#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_already_exec#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("exec_already_exec#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_apply_amt#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("exec_apply_amt#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_already_amt#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("exec_already_amt#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("exec_close_time#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("exec_close_time#" + i)))));

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }

            countCell = value.getColumnLatestCell(FAMILY_DET,
                Bytes.toBytes("guar_count"));
            if (countCell == null)
                System.out.println("---------guar--------");
            if (countCell != null) {
                count = Bytes.toInt(CellUtil.cloneValue(countCell));

                for (int i = 0; i < count; i++) {
                    sb.delete(0, sb.length());
                    sb.append(STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF
                        + STR_TB_PREFIX_FLAG);
                    sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i).append(STR_FILED_SPLIT_FLAG);

                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_deal_time#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("guar_deal_time#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_for_person#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("guar_for_person#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_cert_type#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("guar_cert_type#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_cert_no#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("guar_cert_no#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_bank_name#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("guar_bank_name#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_type#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("guar_type#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_contract_amt#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("guar_contract_amt#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_amount#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("guar_amount#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_deadline#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("guar_deadline#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("guar_balance#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("guar_balance#" + i)))));

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }

            countCell = value.getColumnLatestCell(FAMILY_DET,
                Bytes.toBytes("cred_count"));
            if (countCell == null)
                System.out.println("---------cred--------");
            if (countCell != null) {
                count = Bytes.toInt(CellUtil.cloneValue(countCell));

                for (int i = 0; i < count; i++) {
                    sb.delete(0, sb.length());
                    sb.append(STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF
                        + STR_TB_PREFIX_FLAG);
                    sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i).append(STR_FILED_SPLIT_FLAG);

                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_type#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_type#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_sub_type#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_sub_type#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_class#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_class#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_status#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_status#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_currency#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_currency#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_deal_time#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_deal_time#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_bank_name#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_bank_name#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_period#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toInt(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_period#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_repay_type#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("cred_repay_type#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_balance#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_balance#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_total_balance#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_total_balance#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_close_date#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_close_date#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_deadline_date#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_deadline_date#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_clear_date#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_clear_date#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_overdue_count#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toInt(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_overdue_count#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_overdue_amount#" + i)) == null
                            ? ""
                            : String.valueOf(
                                Bytes.toBigDecimal(value.getValue(FAMILY_DET,
                                    Bytes
                                        .toBytes("cred_overdue_amount#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_overdue_mons#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toInt(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("cred_overdue_mons#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("cred_overdue_90days_mons#" + i)) == null
                            ? ""
                            : String.valueOf(Bytes
                                .toInt(value.getValue(FAMILY_DET, Bytes.toBytes(
                                    "cred_overdue_90days_mons#" + i)))));

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }

            countCell = value.getColumnLatestCell(FAMILY_DET,
                Bytes.toBytes("quer_count"));
            if (countCell == null)
                System.out.println("---------quer--------");
            if (countCell != null) {
                count = Bytes.toInt(CellUtil.cloneValue(countCell));

                for (int i = 0; i < count; i++) {
                    sb.delete(0, sb.length());
                    sb.append(STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF
                        + STR_TB_PREFIX_FLAG);
                    sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                    sb.append(i).append(STR_FILED_SPLIT_FLAG);

                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("quer_no#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("quer_no#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("quer_time#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toLong(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("quer_time#" + i)))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("quer_agent#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("quer_agent#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("quer_reason#" + i)) == null
                            ? ""
                            : Bytes.toString(value.getValue(FAMILY_DET,
                                Bytes.toBytes("quer_reason#" + i))))
                        .append(STR_FILED_SPLIT_FLAG);
                    sb.append(value.getValue(FAMILY_DET,
                        Bytes.toBytes("quer_days#" + i)) == null
                            ? ""
                            : String
                                .valueOf(Bytes.toInt(value.getValue(FAMILY_DET,
                                    Bytes.toBytes("quer_days#" + i)))));

                    this.value.set(sb.toString());
                    context.write(this.key, this.value);
                }
            }

            /*
             * String qualifier = null;
             * String qualifierField = null;
             * String qualifierNum = null;
             * for(Cell cell : value.rawCells()) {
             * if(!Bytes.toString(FAMILY_DET).equals(Bytes.toString(cell.
             * getFamily())))
             * continue;
             * qualifier = Bytes.toString(cell.getQualifier());
             * if(qualifier == null || "".equals(qualifier) ||
             * qualifier.lastIndexOf("#") == -1)
             * continue;
             * qualifierField = qualifier.substring(0,
             * qualifier.lastIndexOf("#")).toLowerCase();
             * qualifierNum = qualifier.substring(qualifier.lastIndexOf("#") +
             * 1);
             * Map<String, byte[]> tmpMap = map.get(qualifierNum);
             * if(tmpMap == null) {
             * tmpMap = new HashMap<String, byte[]>();
             * map.put(qualifierNum, tmpMap);
             * }
             * tmpMap.put(qualifierField, cell.getValueArray());
             * }
             * StringBuffer execSb = new StringBuffer();
             * StringBuffer guarSb = new StringBuffer();
             * StringBuffer credSb = new StringBuffer();
             * StringBuffer querSb = new StringBuffer();
             * execSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF +
             * STR_TB_PREFIX_FLAG);
             * execSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * guarSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF +
             * STR_TB_PREFIX_FLAG);
             * guarSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * credSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF +
             * STR_TB_PREFIX_FLAG);
             * credSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * querSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF +
             * STR_TB_PREFIX_FLAG);
             * querSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * this.key.set(keyStr);
             * for(String mapKey : map.keySet()) {
             * Map<String, byte[]> tmpMap = map.get(mapKey);
             * execSb.delete(0, execSb.length());
             * guarSb.delete(0, execSb.length());
             * credSb.delete(0, execSb.length());
             * querSb.delete(0, execSb.length());
             * execSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF +
             * STR_TB_PREFIX_FLAG);
             * execSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * execSb.append(mapKey).append(STR_FILED_SPLIT_FLAG);
             * credSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF +
             * STR_TB_PREFIX_FLAG);
             * credSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * credSb.append(mapKey).append(STR_FILED_SPLIT_FLAG);
             * guarSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF +
             * STR_TB_PREFIX_FLAG);
             * guarSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * guarSb.append(mapKey).append(STR_FILED_SPLIT_FLAG);
             * querSb.append(STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF +
             * STR_TB_PREFIX_FLAG);
             * querSb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
             * querSb.append(mapKey).append(STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_case_no") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_case_no"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_court") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_court"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_reason") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_reason"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_over_way") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_over_way"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_filing_time") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("exec_filing_time")))).
             * append(STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_status") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_status"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_apply_exec") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_apply_exec"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_already_exec") == null ? "" :
             * Bytes.toString(tmpMap.get("exec_already_exec"))).append(
             * STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_apply_amt") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("exec_apply_amt")))).
             * append(STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_already_amt") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("exec_already_amt")))).
             * append(STR_FILED_SPLIT_FLAG);
             * execSb.append(tmpMap.get("exec_close_time") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("exec_close_time"))));
             * this.value.set(execSb.toString());
             * context.write(this.key, this.value);
             * guarSb.append(tmpMap.get("guar_deal_time") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("guar_deal_time")))).
             * append(STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_for_person") == null ? "" :
             * Bytes.toString(tmpMap.get("guar_for_person"))).append(
             * STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_cert_type") == null ? "" :
             * Bytes.toString(tmpMap.get("guar_cert_type"))).append(
             * STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_cert_no") == null ? "" :
             * Bytes.toString(tmpMap.get("guar_cert_no"))).append(
             * STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_bank_name") == null ? "" :
             * Bytes.toString(tmpMap.get("guar_bank_name"))).append(
             * STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_type") == null ? "" :
             * Bytes.toString(tmpMap.get("guar_type"))).append(
             * STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_contract_amt") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("guar_contract_amt")))).
             * append(STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_amount") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("guar_amount")))).append
             * (STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_deadline") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("guar_deadline")))).append
             * (STR_FILED_SPLIT_FLAG);
             * guarSb.append(tmpMap.get("guar_balance") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("guar_balance"))));
             * this.value.set(guarSb.toString());
             * context.write(this.key, this.value);
             * credSb.append(tmpMap.get("cred_type") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_type"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_sub_type") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_sub_type"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_class") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_class"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_status") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_status"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_currency") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_currency"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_deal_time") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("cred_deal_time")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_bank_name") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_bank_name"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_period") == null ? "" :
             * String.valueOf(Bytes.toInt(tmpMap.get("cred_period")))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_repay_type") == null ? "" :
             * Bytes.toString(tmpMap.get("cred_repay_type"))).append(
             * STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_balance") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("cred_balance")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_total_balance") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("cred_total_balance"))))
             * .append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_close_date") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("cred_close_date")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_deadline_date") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("cred_deadline_date")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_clear_date") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("cred_clear_date")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_overdue_count") == null ? "" :
             * String.valueOf(Bytes.toInt(tmpMap.get("cred_overdue_count")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_overdue_amount") == null ? "" :
             * String.valueOf(Bytes.toDouble(tmpMap.get("cred_overdue_amount")))
             * ).append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_overdue_mons") == null ? "" :
             * String.valueOf(Bytes.toInt(tmpMap.get("cred_overdue_mons")))).
             * append(STR_FILED_SPLIT_FLAG);
             * credSb.append(tmpMap.get("cred_overdue_90days_mons") == null ? ""
             * :
             * String.valueOf(Bytes.toInt(tmpMap.get("cred_overdue_90days_mons")
             * )));
             * this.value.set(credSb.toString());
             * context.write(this.key, this.value);
             * querSb.append(tmpMap.get("quer_no") == null ? "" :
             * String.valueOf(Bytes.toInt(tmpMap.get("quer_no")))).append(
             * STR_FILED_SPLIT_FLAG);
             * querSb.append(tmpMap.get("quer_time") == null ? "" :
             * String.valueOf(Bytes.toLong(tmpMap.get("quer_time")))).append(
             * STR_FILED_SPLIT_FLAG);
             * querSb.append(tmpMap.get("quer_agent") == null ? "" :
             * Bytes.toString(tmpMap.get("quer_agent"))).append(
             * STR_FILED_SPLIT_FLAG);
             * querSb.append(tmpMap.get("quer_reason") == null ? "" :
             * Bytes.toString(tmpMap.get("quer_reason"))).append(
             * STR_FILED_SPLIT_FLAG);
             * querSb.append(tmpMap.get("quer_days") == null ? "" :
             * String.valueOf(Bytes.toInt(tmpMap.get("quer_days"))));
             * this.value.set(querSb.toString());
             * context.write(this.key, this.value);
             * }
             */
        }

        private Object[] decodeRowkey(byte[] raw) {
            if (raw.length < SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
                return null;
            }
            Object[] ret = new Object[2];
            ret[0] = Long.MAX_VALUE
                - Bytes.toLong(raw, SIZEOF_MD5, Bytes.SIZEOF_LONG);

            if (raw.length > SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
                ret[1] = Bytes.toString(raw, SIZEOF_MD5 + Bytes.SIZEOF_LONG);
            }

            return ret;
        }
    }

    private static class CreditReportsReducer
            extends Reducer<Text, Text, Text, Text> {
        private NullWritable key2 = NullWritable.get();
        private MultipleOutputs<Text, Text> mos = null;

        protected void setup(Context context)
                throws IOException, InterruptedException {
            mos = new MultipleOutputs<Text, Text>(context);
        }

        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            mos.close();
        }

        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text value : values) {
                String[] contents = value.toString().split(STR_TB_PREFIX_FLAG);
                if (contents == null)
                    continue;

                if (STR_TARGET_POSITION_CREDIT_REPORTS_JOB_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_JOB_INF,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_CREDIT_REPORTS_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_INF, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF,
                        key2, new Text(contents[1]));
                }
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }
        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum", "datanode1,datanode2,datanode3");
        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");

        /*
         * conf.set("param.out.path",
         * "hdfs://10.10.15.10:8020/tmp/output_xxh_rongreport/");
         * conf.set("param.start_date", "2017-09-19");
         * conf.set("param.stop_date", "2017-09-20");
         */
        conf.set("param.out.path", args[0]);
        conf.set("param.start_date", args[1]);
        conf.set("param.stop_date", args[2]);

        String tableName = "credit_reports";
        String outputDir = conf.get("param.out.path");
        System.out.println("开始日期:" + conf.get("param.start_date"));
        System.out.println("结束日期:" + conf.get("param.stop_date"));
        System.out.println("输入HBase表名:" + tableName);
        System.out.println("输出路径:" + outputDir);

        Long startTime = DateUtil
            .formateToTimestamp(conf.get("param.start_date"));
        Long endTime = DateUtil.formateToTimestamp(conf.get("param.stop_date"));

        Path outpath = new Path(outputDir);
        FileSystem fs = outpath.getFileSystem(conf);
        if (fs.exists(outpath)) {
            fs.delete(outpath, true);
        }

        job.setJarByClass(RongReportsExtractor.class);
        Scan scan = new Scan();

        scan.addFamily(FAMILY_INF);
        scan.addFamily(FAMILY_DET);
        scan.addFamily(FAMILY_JOB);

//		scan.setCacheBlocks(false);
//		scan.setCaching(200);
//		scan.setBatch(100);
        scan.setMaxVersions(1);
//		scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName));
        scan.setTimeRange(startTime, endTime + 14400000);

        TableMapReduceUtil.initTableMapperJob(tableName, scan,
            CreditReportsMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, outpath);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_JOB_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_EXEC_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_GUAR_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_CRED_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_CREDIT_REPORTS_QUER_INF, TextOutputFormat.class,
            Text.class, Text.class);

        job.setReducerClass(CreditReportsReducer.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        System.exit(ToolRunner.run(HBaseConfiguration.create(),
            new CreditReportsExtractor(), args));
    }
}
